pyspark 에서 발생한 문제다. 이 버그에 하루를 날렸다.
for ds in dds:
my_rdd[ds] = sc.textFile(my_path[ds]).map(lambda line: my_map(line, ds)이 소스는 문제가 된다. dds = [3, 4, 5, 6] 이라고 치자. 이때 my_map에는 6만 전달된다. my_rdd와 my_path는 문제가 없는데 my_map이 문제다.
추정하면 이렇다. spark의 많은 함수들은 copy on write 혹은 lazy loading 방식으로 실행된다. 뒤에서 실제로 쓰는 컨텍스트가 아니면 건너뛰기조차 하는 똘똘한 설계다.
위의 for 루프 역시 실행은 미루고 와꾸만 잡아둔다. 이 때 map안에 있는 my_map을 셋팅하면서 ds값을 실제로 넘기지 않고, 현대적인 언어들이 그렇듯이 레퍼런스 (주소)만 넘길 것이다. 내부의 코드를 실제로 실행하는 것은 미뤄둔다. 자 이제 for 루프가 다 돌고나면 ds는 dds 배열의 최종 값을 포인팅 하고 있을 것 같다. 파이썬 메커니즘을 알지 못하기에 ~같다만 씁니다만 뭐 대충 그렇겠죠.
my_rdd를 직접 쓸 시간이 오면 copy on write가 발동한다. 이제 my_map을 계속 호출하는데 아까 ds라고 받은 레퍼런스는 6만 찍고 있다. 이제 모든 my_rdd[3], my_rdd[...] 들은 my_map(line, 6)만 사용하여 일을 한다. 망한 거다. 하루 홀랑.
C언어로 따지면 ds는 포인터이고 아규먼트에서 전달시에는 2중 포인터로 전달 됐을 것이다. 아규먼트에서 사용한 포인터를 pds라고 치면 pds가 가리키고 있는 ds 자체의 주소는 변하지 않는다. ds 자체의 주소에 들어있는 "값"은 3, 4, 5, 6의 주소를 차례차례 가리키고 있고 최종적으로 6이 들어있는 주소를 가리킨다. my_map은 pds를 받았고 pds의 값은 변함없이 ds를 가리키고 있는데 ds가 가리키는 주소는 for 루프가 돌고나서 6이 들어있는 주소만 가리킨다. 이게 말이야 똥이야 2중 포인터가 이렇게 위험합니다 여러분!!
spark 버그라고 하긴 그렇고 여튼 이런 문제가 있다. 추정은 추정이므로 참고만 하세용.
+) 내용추가 : 검색해보면 이것을 spark의 외부변수 참조 문제라고 명명되어 있다. 예컨데 이렇다.
실제로는 lazy evaluation 문제라고도 부르고 (최후에 evaluation 된 value가 반영됨, 함수형 프로그래밍에서는 이렇다고) 하는데, 나처럼 로우엔드에서 해석하는 사람은 별로 없는 듯. 하지만 사실 같은 이야기라고 볼 수 있다. spark map에서 path를 가져오려면 inputformatter를 상속받아 꾸미는 게 방법인데 pyspark에서는 마땅한 방법이 없는 듯 하다.
+) 그냥 이렇게 해결하자.
class ds_rdd:
def __init__(self, ds):
self.ds = ds
def get_ds(self):
ds = self.ds
return sc.textFile(','.join(tds[ds])).map(lambda line: map_ds0(line, ds))
if 1:
data_ds = {}
for ds in tds:
data_ds[ds] = ds_rdd(ds).get_ds()
print data_ds[64].take(1)
exit(0)